package org.budo.storm.trident.operation;

import java.util.Map;

import org.apache.storm.trident.operation.Filter;
import org.apache.storm.trident.operation.TridentOperationContext;
import org.apache.storm.trident.tuple.TridentTuple;
import org.budo.support.spring.context.aware.BudoApplicationContextAware;
import org.budo.support.spring.util.SpringUtil;
import org.springframework.context.support.ClassPathXmlApplicationContext;

import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.Setter;

/**
 * @author lmw
 */
@Getter
@Setter
@NoArgsConstructor
@AllArgsConstructor
public class RuntimeFilterReference implements Filter {
    private static final long serialVersionUID = -5274642655566730557L;

    private String beanName;

    private String configLocation;

    private Filter getFilter() {
        if (null != BudoApplicationContextAware.getInstance()) {
            return (Filter) SpringUtil.getBean(this.getBeanName());
        }

        synchronized (RuntimeFilterReference.class) {
            if (null != BudoApplicationContextAware.getInstance()) {
                return (Filter) SpringUtil.getBean(this.getBeanName());
            }

            new ClassPathXmlApplicationContext(this.getConfigLocation());
            return (Filter) SpringUtil.getBean(this.getBeanName());
        }
    }

    @Override
    public void prepare(Map conf, TridentOperationContext context) {
        this.getFilter().prepare(conf, context);
    }

    @Override
    public void cleanup() {
        this.getFilter().cleanup();
    }

    @Override
    public boolean isKeep(TridentTuple tuple) {
        return this.getFilter().isKeep(tuple);
    }
}